package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2WordCountOnDSL {
  def main(args: Array[String]): Unit = {
    /**
     * 1 创建spark sql 环境
     */
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("wc")
      .config("spark.sql.shuffle.partitions", 1) //指定spark sql shuffle之后rdd的分区数
      .getOrCreate()

    val linesDF: DataFrame = spark
      .read
      .format("csv") // 读取数据的格式
      .option("sep", "|") // 字段的分隔符
      .schema("line STRING") // 指定字段名和字段类型
      .load("data/words.txt") //指定读取数据的路径

    /**
     * 使用DSL统计单词的数量
     * DSL:类SQL语法,介于代码和SQL中间的写入
     *
     * $"line": 获取列对象
     *
     */
    //导入spark sql的函数包
    import org.apache.spark.sql.functions._
    //导入spark sql的隐式转换
    import spark.implicits._

    val resultDF: DataFrame = linesDF
      //一行转换成多行
      .select(explode(split($"line", ",")) as "word")
      //按照单词进行分组
      .groupBy($"word")
      //统计单词的数量
      .agg(count($"word") as "num")

    //保存结果
    resultDF
      .write
      .format("json")
      .mode(SaveMode.Overwrite)
      .save("data/word_count1")
  }
}
